Skip to content

iceoryx2 事件机制:一次性通知与数据控制流分离 ​

发布订阅解决「传什么数据」,但很多场景只需要一个信号:电量低了、撞墙了、传感器掉了。用 pub-sub 传这类信号,要么发空消息、要么让接收方高频轮询,都是绕路。iceoryx2 的事件(Event)机制专为「通知状态变化」设计:一次性信号、不排队、不丢。本文介绍事件模型的核心语义、C++ 双端实现,以及它与发布订阅组合出的「数据流与控制流分离」模式,适合正在使用或评估 iceoryx2 的开发者。

状态与事件 ​

先分清两个概念:

  • 状态(State):内存中的当前事实,例如电池当前是低电量还是正常;
  • 事件(Event):状态发生了变化这一通知本身。

事件机制由两种端口组成:通知器(Notifier) 发送通知,监听器(Listener) 等待通知。监听器平时休眠,只在真正需要关注的事件发生时被唤醒。

核心语义:不排队、会合并、顺序未定义 ​

这是理解事件机制最关键的部分,它和消息队列是两种模型:

  • 不排队:事件不像 pub-sub 消息那样在缓冲区里排队等待消费。每个事件 ID 更接近一个带锁存的状态标志位;
  • 会合并:如果通知器在监听器处理之前连续发送了多次同一事件,监听器醒来只会感知到一次「该状态变过」;
  • 顺序未定义:多个事件到达的先后顺序没有保证,监听器只知道「至少有某状态变化发生过」;
  • 不丢失:已发出的事件在监听器观察到之前不会丢失——综上,语义是「每个事件 ID 至少被通知一次,可见次数至多一次」。

这套语义适合「电平触发」型条件:只关心当前事实(电池是否低),不关心它被报了几次。如果需要每条消息都独立送达,那应该用 pub-sub 而不是事件。

双端实现 ​

通知器端 ​

创建节点与事件服务,用 EventId 区分不同类型的事件:

cpp
#include "iox2/iceoryx2.hpp"

using namespace iox2;

auto node = NodeBuilder().create<ServiceType::Ipc>().value();

auto service = node.service_builder(ServiceName::create("system_health_events").value())
                   .event()
                   .open_or_create()
                   .value();

auto notifier = service.notifier_builder().create().value();

// 为不同类型的事件分配唯一标识
const auto wall_was_hit = EventId(0);
const auto battery_is_low = EventId(1);

while (node.wait(iox2::bb::Duration::from_secs(1)).has_value()) {
    if ( bump_sensor_was_activated() ) {
        notifier.notify_with_custom_event_id(wall_was_hit).value();
    }

    if ( battery_state() < battery_threshold ) {
        notifier.notify_with_custom_event_id(battery_is_low).value();
    }
}

监听器端 ​

打开同一个事件服务,阻塞等待并在回调里按事件 ID 分发处理:

cpp
#include "iox2/iceoryx2.hpp"

using namespace iox2;

auto node = NodeBuilder().create<ServiceType::Ipc>().value();

auto service = node.service_builder(ServiceName::create("system_health_events").value())
                   .event()
                   .open_or_create()
                   .value();
auto listener = service.listener_builder().create().value();

while (listener.blocking_wait(
    [](auto event) {
        if ( event.id() == battery_is_low ) {
            activate_battery_warning_light();
        }

        if ( event.id() == wall_was_hit ) {
            go_into_parking_position();
        }
    }).has_value()) {}

注意 blocking_wait 的回调参数是事件对象,取标识用 event.id()。

事件驱动模式:数据流与控制流分离 ​

单一模式之外,iceoryx2 官方示例展示了一个更有价值的组合思路:

  • 数据流描述「函数接收什么输入」——由发布订阅、请求响应承载;
  • 控制流描述「函数何时执行」——由事件承载。

两者分离后,不需要用「发空消息」或「过早唤醒」这类 hack 来触发执行,还能支持无数据唤醒和多输入等待(所有输入到齐前不中断)。

以机器人紧急刹车系统为例:

  • 选择性响应:只在障碍物接近时刹车,而不是每个数据样本都触发一轮处理;
  • 趋势分析:刹车决策需要最近 3 个距离样本(位置、速度、加速度);
  • 容错:传感器死亡时进入安全状态(停车);刹车后重新启动时仍能取到历史数据。

发布者:pub-sub 传数据 + 事件传信号 ​

cpp
auto node = NodeBuilder().create<ServiceType::Ipc>().value();

// 1. 发布订阅服务(距离数据流)
auto pubsub_service = node.service_builder(
                        ServiceName::create("distance_to_obstacle").value())
                   .publish_subscribe<Distance>()
                   .subscriber_max_buffer_size(3)
                   .history_size(3)
                   .subscriber_max_borrowed_samples(3)
                   .open_or_create()
                   .value();

// 2. 事件服务(控制信号)
auto ultra_sonic_service_dead = EventId(10);
auto event_service = node.service_builder(
                        ServiceName::create("distance_to_obstacle").value())
                   .event()
                   .notifier_dead_event(ultra_sonic_service_dead)
                   .open_or_create()
                   .value();

auto publisher = pubsub_service.publisher_builder().create().value();
auto notifier = event_service.notifier_builder().create().value();
auto obstacle_too_close = EventId(5);

while (node.wait(iox2::bb::Duration::from_millis(100)).has_value()) {
    auto sample = publisher.loan_uninit().value();

    auto distance = get_ultra_sonic_sensor_distance();
    auto initialized_sample = sample.write_payload(Distance { distance, 42.0 });

    if (distance < distance_threshold) {
        notifier.notify_with_custom_event_id(obstacle_too_close).value();
    }
}

几项配置的含义:

  • subscriber_max_buffer_size(3):订阅者最大缓冲区大小;
  • history_size(3):历史样本保留数量——新订阅者或重启的处理方能立即拿到最近 3 个样本;
  • subscriber_max_borrowed_samples(3):同时可借用的样本数上限;
  • notifier_dead_event(ultra_sonic_service_dead):把某个 EventId 注册为「通知方死亡」信号,依托 iceoryx2 内置的参与者健康监控自动触发。

订阅者:被事件唤醒,醒来再取数据 ​

订阅者不再按 100ms 轮询,而是阻塞在事件上。被唤醒后,如果是「障碍物过近」,一次性取走缓冲区里的最近 3 个样本做趋势分析;如果是「传感器死亡」,直接进安全状态:

cpp
while (listener.blocking_wait(
    [](auto event) {
        // 传感器死亡事件
        if ( event.id() == ultra_sonic_service_dead ) {
            go_into_parking_position();
        }

        // 障碍物过近事件
        if ( event.id() == obstacle_too_close ) {
            std::vector<Sample<ServiceType::Ipc, Distance, void>> last_samples;

            auto sample = subscriber.receive().value();
            while (sample.has_value()) {
                last_samples.push_back(std::move(sample.value()));
                if (last_samples.size() == 3) break;
                sample = subscriber.receive().value();
            }

            perform_break(last_samples);
        }
    }).has_value()) {}

数据与控制两条通路各司其职:pub-sub 负责持续的距离样本流并保留历史,事件负责「现在必须处理」的触发时机。

要点 ​

  • 事件是锁存的状态标志,不是消息队列:会合并、顺序未定义、每个事件 ID 至少通知一次
  • 需要每条消息独立送达时用 pub-sub,需要「状态变了叫我」时用事件
  • EventId 区分事件类型,blocking_wait 回调里用 event.id() 分发
  • 数据流与控制流分离是事件模式的正确打开方式:pub-sub 管数据与历史,事件管触发时机
  • notifier_dead_event 借助内置健康监控实现参与者死亡检测,容错场景直接可用
最近更新

基于 VitePress 构建